package cn.com.dhcc.ddi.task.service.extract;

import cn.com.dhcc.app.core.exception.DaoException;
import cn.com.dhcc.app.core.exception.NiceException;
import cn.com.dhcc.app.core.exception.ServiceException;
import cn.com.dhcc.app.core.handler.compresshandler.CompressResult;
import cn.com.dhcc.app.core.handler.compresshandler.CompressService;
import cn.com.dhcc.app.core.handler.compresshandler.CompressServiceFactory;
import cn.com.dhcc.app.core.handler.compresshandler.CompressStatic;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.tag.CommonStaticCode;
import cn.com.dhcc.app.core.util.UUIDGenerator;
import cn.com.dhcc.app.pub.core.consts.DdiProperties;
import cn.com.dhcc.app.pub.core.tag.staticcode.StaticCode;
import cn.com.dhcc.ddi.datamapping.dao.DataMappingDao;
import cn.com.dhcc.ddi.datamapping.vo.DataMapping;
import cn.com.dhcc.ddi.datasource.dao.DataSourceDao;
import cn.com.dhcc.ddi.datasource.vo.DataSourceVo;
import cn.com.dhcc.ddi.datasource.vo.TableColVo;
import cn.com.dhcc.ddi.datasource.vo.TableInfoVo;
import cn.com.dhcc.ddi.task.dao.GatherChangeDbDateDao;
import cn.com.dhcc.ddi.task.dao.TableLogDao;
import cn.com.dhcc.ddi.task.dao.TaskDao;
import cn.com.dhcc.ddi.task.dao.TaskExecuteLogDao;
import cn.com.dhcc.ddi.task.service.ChangeVo;
import cn.com.dhcc.ddi.task.service.Result;
import cn.com.dhcc.ddi.task.service.TaskTabService;
import cn.com.dhcc.ddi.task.vo.TableLog;
import cn.com.dhcc.ddi.task.vo.Task;
import cn.com.dhcc.ddi.task.vo.TaskExecuteLog;
import org.apache.commons.collections.ListUtils;
import org.apache.commons.io.FileUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.File;
import java.io.IOException;
import java.util.*;

/**
 * 数据库数据抽取抽象类
 * @Date：2016/11/18 9:17
 * @author：张洪涛
 * @版权所有：东华软件
 * @版本：1.0
 */
@Service
public abstract class EDataSourceDBAbstract implements EDataSourceInterface {
    private static Logger logger = Log.getLogger(Log.LOGTYPE.DDI);
    @Autowired
    private TaskExecuteLogDao taskExecuteLogDao;
    @Autowired
    private TableLogDao tableLogDao;
    @Autowired
    private TaskDao taskDao;
    @Autowired
    private DataMappingDao dmDao;
    @Autowired
    private DataSourceDao dsDao;
    @Autowired
    private GatherChangeDbDateDao changeDbDateDao;
    @Autowired
    private GatherChangeDbData gatherChangeDbData;
    @Autowired
    private TaskTabService taskTabService;

    /**
     *生成数据文件
     * @param uploadRoot
     * @return
     * @throws ServiceException
     */
    private File createDataXmlFile(String uploadRoot) throws ServiceException{
        File uploadFolder = new File(uploadRoot);
        if (!uploadFolder.exists()) {
            logger.info(uploadFolder + "目录不存在，创建。");
            uploadFolder.mkdirs();
        }
        String dataXmlPath = uploadRoot+File.separator+ UUIDGenerator.getUUID()+".xml";
        File dataXmlFile = new File(dataXmlPath);
        if (!dataXmlFile.exists()) {
            try {
                logger.info(dataXmlFile + "不存在，创建新文件。");
                dataXmlFile.createNewFile();
            } catch (IOException e) {
                logger.error("创建文件" + dataXmlFile + "发生异常：",e);
                throw new ServiceException(e);
            }
        }
        return dataXmlFile;
    }

    /**
     *数据文件写入数据
     * @param dataXmlFile
     * @param content
     * @throws ServiceException
     */
    private void writeData(File dataXmlFile, String content) throws ServiceException{
        try {
            FileUtils.writeStringToFile(dataXmlFile, content, "utf-8",true);
        } catch (IOException e) {
            logger.error("写入data发生异常",e);
            throw new ServiceException("写入data发生异常",e);
        }
    }

    /**
     * 计算数据总数
     * @param sqlNum
     * @return
     */
    private long computeSqlNum(Map<String, List<Integer>> sqlNum){
        long sum = 0;
        if(sqlNum == null){
            return sum;
        }
        for (List<Integer> nums : sqlNum.values()) {
            for (Integer num : nums) {
                sum += num;
            }
        }
        return sum;
    }

    /**
     * 压缩文件
     * @param uploadRoot
     * @param before
     * @return
     * @throws ServiceException
     */
    private CompressResult<String> compressFile(String uploadRoot, File before) throws ServiceException{
        try {
            CompressService compressService = CompressServiceFactory.createCompressService(CompressStatic.COMPRESS_ALGORITHM.JAVA_GZIP);
            String after = uploadRoot+File.separator+before.getName()+".edata";//压缩后文件名
            CompressResult<String> sresult = compressService.compress(before,after,true);
            logger.info("压缩文件成功:" + sresult.getCommpressResultObj());
            return sresult;
        } catch (Exception e1) {
            logger.error("压缩文件" + before + "发生异常",e1);
            throw new ServiceException("压缩文件" + before + "发生异常",e1);
        }
    }
    /**
     * 生成执行详情
     * @param changeVo
     * @return
     */
    private String  createContentDetail(ChangeVo changeVo) {
        logger.info("进入方法createCendContentDetail().");
        StringBuilder rs = new StringBuilder();
        if(changeVo.isTableDataFlag()){
            int insertSqlNum = 0;
            StringBuilder tables = new StringBuilder();
            for (String tableName : changeVo.getTableDataNum().keySet()) {
                List<Integer> nums =  changeVo.getTableDataNum().get(tableName);
                int count = 0;
                for (Integer num : nums) {
                    insertSqlNum += num;
                    count += num;
                }
                String startValue = changeVo.getStartValue().get(tableName) != null && !"".equals(changeVo.getStartValue().get(tableName)) ? changeVo.getStartValue().get(tableName): "";
                String endValue = changeVo.getEndValue().get(tableName) != null && !"".equals(changeVo.getEndValue().get(tableName)) ? changeVo.getEndValue().get(tableName): "";
                tables.append("表：").append(tableName).append(",记录数：").append(count).append(",抽取增量值:").append(startValue ).append(" - ").append(endValue).append("\r\n");
            }
            rs.append("数据抽取结果,总数:").append(insertSqlNum).append("\r\n").append(tables);
        }
        logger.info("执行结束createCendContentDetail().返回值：" + rs.toString());
        return rs.toString();
    }

    /**
     *
     * @param task
     * @param currentTimeSql
     * @param dataXmlPath
     * @return
     * @throws NiceException
     * @throws DaoException
     * @throws ServiceException
     */
    private ChangeVo getChangeSqlList(Task task, String taskLogId, String dataXmlPath) throws NiceException, DaoException, ServiceException {
        logger.info("进入方法getChangeSqlList()。");
        String taskId = task.getId();
        ChangeVo changeVo = new ChangeVo();
        changeVo.setExecuteTableNames(new HashSet<String>());
        changeVo.setTableLogs(new HashMap<String, TableLog>());
        //1根据任务获取接收系统和每个接收系统本次任务需要同步的表和字段
        List<TableInfoVo> tasksTables = taskTabService.getListBytaskId(taskId);
        //根据主外键关系开始排序
        tasksTables = sortTableByFk(tasksTables);

        DataMapping dm = dmDao.get(task.getMpId());
        //根据数据库类型获取数据库抽取的实现
        DataSourceVo dataSource = dsDao.get(dm.getSrcDsId());

        //获取数据
        Map<String, TableLog> tableName_tableLog = changeDbDateDao.getTableLogBySysId(task.getId(),tasksTables);
        if(tableName_tableLog == null){
            tableName_tableLog = new HashMap<String, TableLog>();
        }
        //表同步日志记录
        ChangeVo tempVo = gatherChangeDbData.gatherChangeSql(task,dataSource,tasksTables,tableName_tableLog,taskLogId,dataXmlPath);
        if(tempVo.isTableDataFlag()){
            changeVo.setTableDataNum(tempVo.getTableDataNum());
            changeVo.setTableDataFlag(tempVo.isTableDataFlag());
            changeVo.getExecuteTableNames().addAll(tempVo.getExecuteTableNames());
            changeVo.setStartValue(tempVo.getStartValue());
            changeVo.setEndValue(tempVo.getEndValue());
        }
        if(tableName_tableLog!=null){
            changeVo.getTableLogs().putAll(tableName_tableLog);
        }

        logger.info("执行完毕getChangeSqlList().");
        return changeVo;
    }


    /**
     * 根据外键关系排序
     * @param tasksTables
     * @return
     */
    private List<TableInfoVo> sortTableByFk(
            List<TableInfoVo> tasksTables) {
        logger.info("进入方法sortTableByFk()。");
        if (tasksTables==null || tasksTables.isEmpty()) {
            return tasksTables;
        }
        List<TableInfoVo> newList=new ArrayList<TableInfoVo>();
        for (TableInfoVo taskTab : tasksTables) {
            if (getTables(tasksTables, taskTab).isEmpty() && !newList.contains(taskTab)) {
                newList.add(taskTab);
            }
        }

        for (TableInfoVo taskTab : newList) {
            if (newList.contains(taskTab)) {
                tasksTables.remove(taskTab);
            }
        }
        sortTab(tasksTables, newList);
        logger.info("执行完毕sortTableByFk()。");
        return newList;
    }

    /**
     * 从tasksTables集合中，找出tab表外键关联的表的集合
     * @param tasksTables
     * @param tab
     * @return
     */
    private List<TableInfoVo> getTables(List<TableInfoVo> tasksTables, TableInfoVo tab){
        List<TableInfoVo> tabs=new ArrayList<TableInfoVo>();
        List<TableColVo> cols=tab.getTableColVos();
        for (TableColVo taskTabCol : cols) {
            if (taskTabCol.getFkTableName()!=null && !"".equals(taskTabCol.getFkTableName())) {//某个表的外键表
                TableInfoVo tab1=getTable(tasksTables, taskTabCol);
                if (tasksTables.contains(tab1)) {
                    tabs.add(tab1);
                }
            }
        }
        return tabs;
    }


    /**
     *
     * @param tasksTables
     * @param pkCol
     * @return
     */
    private TableInfoVo getTable(List<TableInfoVo> tasksTables, TableColVo pkCol){
        for (TableInfoVo taskTab : tasksTables) {
            if (pkCol.getFkTableName().equals(taskTab.getTableName())) {
                return taskTab;
            }
        }
        return null;
    }

    /**
     * 排序
     * @param tasksTables 有主键的列表
     * @param newList  排序的列表
     * @return
     */
    @SuppressWarnings("unchecked")
    private List<TableInfoVo> sortTab(List<TableInfoVo> tasksTables, List<TableInfoVo> newList){
        if (tasksTables==null || tasksTables.isEmpty()) {
            return newList;
        }
        while (tasksTables.size()>0) {
            if (tasksTables.size()==1) {
                newList.add(tasksTables.get(0));
            }else if(tasksTables.size()>1){
                callback(tasksTables, tasksTables.get(0), newList);
            }
            tasksTables= ListUtils.subtract(tasksTables, newList);
        }
        return newList;
    }

    /**
     * 回调函数，取出没有外键的表，放入到newList中去
     * @param tasksTables
     * @param taskTab
     * @param newList
     */
    public void callback(List<TableInfoVo> tasksTables,TableInfoVo taskTab,List<TableInfoVo> newList){
        List<TableInfoVo> tabs = getTables(tasksTables, taskTab);
        if (tabs.isEmpty() && !newList.contains(taskTab)) {
            newList.add(taskTab);
        }else{
            for (TableInfoVo tab : tabs) {
                callback(tasksTables, tab, newList);
                break;
            }
        }
    }


    /**
     * 抽取数据
     * @param task 任务
     * @param taskLogId 任务
     * @param currentTimeSql 获取当前数据库时间（可能做为增量值）
     * @return 返回抽取环节日志
     * @throws ServiceException
     */
    public Result extract(Task task, String taskLogId, String currentTimeSql) throws ServiceException {
        {
            Result result = new Result();
            result.setTaskLogId(taskLogId);
            logger.info("进入方法extract()。");
            int errorTaskNum = 0;
            int successNum = 0;
            ChangeVo changeVo = null;
            long total = 0;//记录抽取sql条数
            try {
                //记录抽取环节日志
                TaskExecuteLog extractLog = new TaskExecuteLog();
                extractLog.setId(UUIDGenerator.getUUID());
                extractLog.setTaskId(task.getId());
                extractLog.setLinkType(TaskExecuteLog.LINK_TYPE.EXTRACT.getCode());
                extractLog.setLinkStartTime(new Date());
                extractLog.setMtime(new Date());
                extractLog.setGroupId(taskLogId);
                taskExecuteLogDao.insert(extractLog);
                result.setLinkLogId(extractLog.getId());
                result.setTaskLogId(extractLog.getGroupId());
                boolean taskExecuteSuccess;
                String uploadRoot = DdiProperties.APP_UPLOAD_FILEDATA_PATH + File.separator + "extract" + File.separator + task.getId() + File.separator+ taskLogId;
                //创建数据文件
                File dataXmlFile = createDataXmlFile(uploadRoot);
                String dataXmlPath = dataXmlFile.getAbsolutePath();
                //写入数据头

                //获取待发送的sql语句
                String resultMsg = "";
                //任务的日志
                TaskExecuteLog taskExecuteLog  = taskExecuteLogDao.get(taskLogId);
                try {
                    //当前增量值
                    writeData(dataXmlFile, "<data>");
                    changeVo = getChangeSqlList(task,taskLogId,dataXmlPath);
                    //记录抽取sql总数
                    total = computeSqlNum(changeVo.getTableDataNum());
                    logger.info("任务"+task.getId()+"数据抽取成功.");
                    taskExecuteSuccess = true;
                } catch (Exception e) {
                    logger.error("数据抽取失败",e);
                    taskExecuteSuccess = false;
                    resultMsg = "数据抽取失败，"+e.getMessage();
                    extractLog.setLogMsg("数据抽取失败");
                }

                boolean hasSqlTosend = false;
                if(taskExecuteSuccess){
                    if(changeVo.isTableDataFlag()){
                        hasSqlTosend = true;
                        //写入结尾
                        writeData(dataXmlFile, "</data>");
                        //压缩
                        CompressResult<String> sresult  = compressFile(uploadRoot, dataXmlFile);
                        String absolutePath = sresult.getCommpressResultObj();
                        File afterFile = new File(absolutePath);
                        //记录抽取环节日志
                        extractLog.setLinkDataFilePath(afterFile.getAbsolutePath());
                        extractLog.setLogMsg("数据抽取成功");
                        //发送数据日志
                        resultMsg = createContentDetail(changeVo);
                        extractLog.setTotal(total);
                        taskExecuteLog.setTotal(total);
                        taskExecuteLogDao.update(taskExecuteLog);
                    }else{//如果没有sql语句要执行
                        resultMsg = "本次任务没有交换数据.";
                        taskExecuteSuccess = true;
                        if(task.getTaskFreqType().equalsIgnoreCase(StaticCode.TASK_FREQ_TYPE.ONETIME.getCode())){
                            task.setStatus(StaticCode.TASK_STATUS.OVER.getCode());
                        }else{
                            task.setStatus(StaticCode.TASK_STATUS.WAIT.getCode());
                        }

                    }
                    result.setStatus(CommonStaticCode.SUCCESSFAIL.SUCCESS.getStatus());
                    extractLog.setLogStatus(CommonStaticCode.SUCCESSFAIL.SUCCESS.getStatus());
                    dataXmlFile.getAbsoluteFile().delete();
                }else{
                    result.setStatus(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus());
                    extractLog.setLogStatus(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus());
                }

                if(hasSqlTosend){
                    //日志回滚
                    if(!taskExecuteSuccess && changeVo !=null && changeVo.getTableLogs()!=null){
                        for (TableLog tableLog : changeVo.getTableLogs().values()) {
                            tableLog.setTempLastValue(null);
                            tableLogDao.update(tableLog);
                        }
                    }
                }
                //更新执行日志
                extractLog.setLinkSummary(resultMsg);
                extractLog.setLinkEndTime(new Date());
                taskExecuteLogDao.update(extractLog);
                //修改任务表状态
                if(!taskExecuteSuccess){
                    task.setStatus(StaticCode.TASK_STATUS.ERROR.getCode());
                }
                taskDao.update(task);
                if(!taskExecuteSuccess){
                    errorTaskNum++;
                }
                if(taskExecuteSuccess&&hasSqlTosend){
                    successNum++;
                }
                return result;
            } catch (Exception e) {
                errorTaskNum++;
                if(!(e instanceof ServiceException)){
                    logger.error(e);
                }
                //中间过程异常，回滚
                if(changeVo !=null&& changeVo.getTableLogs()!=null){
                    try {
                        if(task.getStatus()==null||!task.getStatus().equalsIgnoreCase(StaticCode.TASK_STATUS.OVER.getCode())){
                            logger.info("任务执行状态修改为失败");
                            task.setStatus(StaticCode.TASK_STATUS.ERROR.getCode());
                            taskDao.update(task);
                        }
                        for (TableLog tableLog : changeVo.getTableLogs().values()) {
                            tableLog.setTempLastValue(null);
                            tableLogDao.update(tableLog);
                        }
                    } catch (DaoException e1) {
                        logger.error("日志回滚失败",e1);
                        //回滚失败，不再跑出异常
                    }
                }
            }
            logger.info("执行完毕extract,失败：" + errorTaskNum + ";成功：" + successNum);

        }
        return null;
    }

    public void reciveNotify(String taskLogId,String status,String msg) {
        logger.info("进入方法reciveNotify()。");
        try {
            TaskExecuteLog taskExecuteLog = taskExecuteLogDao.get(taskLogId);
            Task task =  taskDao.get(taskExecuteLog.getTaskId());
            boolean sendSuccess = false;
            if(CommonStaticCode.SUCCESSFAIL.SUCCESS.getStatus().equals(status)){
                sendSuccess = true;
            }
            //修改主任务的状态

            //修改数据发送日志状态，改为发送成功

            if(sendSuccess){
                taskExecuteLog.setLogStatus(CommonStaticCode.SUCCESSFAIL.SUCCESS.getStatus());
                taskExecuteLog.setLogMsg("");
            }else{
                logger.info("执行失败，下次重发。");
                taskExecuteLog.setLogMsg(msg);
                taskExecuteLog.setLogStatus(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus());
                task.setStatus(StaticCode.TASK_STATUS.ERROR.getCode());
            }


            //修改表日志
            if (sendSuccess) {
                List<TableLog> tableLogs = tableLogDao.listByTaskId(task.getId());
                for (TableLog tableLog : tableLogs) {
                    String lastValue = tableLog.getTempLastValue();
                    //设置上次同步增量值
                    if (!lastValue.isEmpty()){
                        tableLog.setLastValue(lastValue);
                        tableLog.setTempLastValue(null);
                        int uNum  = tableLogDao.update(tableLog);
                        logger.info("updateTaskLogStatus-->actionInCircle更新t_ddi_task_tab_log行数：" + uNum);
                    }
                }
            }
            taskDao.update(task);
            taskExecuteLog.setLinkEndTime(new Date());
            taskExecuteLogDao.update(taskExecuteLog);
        } catch (DaoException e) {
            logger.error(e);
        }
        logger.info("执行完毕reciveNotify()。");
    }

}
